package cfengMiddleware.server.config;

import cfengMiddleware.server.rabbitmq.consumer.BasicConsumer;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @author Cfeng
 * @date 2022/9/15
 * rabbitmq自定义配置文件，需要配置单一消费者工厂，多消费者工厂、消息发送Template对象
 */

@Slf4j
@Configuration
@RequiredArgsConstructor
@EnableRabbit
public class RabbitmqConfig {

    //自动装配的连接工厂,连接到rabbitMQ服务器
    private final CachingConnectionFactory connectionFactory;

    //自动装配的简单Rabbit消息监听器所在容器工厂的配置对象
    private final SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

    //方便快捷代替配置文件，Environment可以方便获取yml文件中的配置项
    private final Environment environment;


    /**
     * 单一消费者实例监听器容器工厂
     */
    @Bean(name = "singleListenerContainer")
    public SimpleRabbitListenerContainerFactory listenerContainerFactory() {
        //消费者监听器所在容器工厂
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        //设置连接的实例
        factory.setConnectionFactory(connectionFactory);
        //设置消息的阐述格式，JSON
        factory.setMessageConverter(new Jackson2JsonMessageConverter());
        //设置并发消费者实例最大数量（Security的并发最大）
        factory.setMaxConcurrentConsumers(1);
        //设置并发消费者实例每个实例拉取的消息数量 1
        factory.setPrefetchCount(1);

        return factory;
    }

    /**
     * 多消费者实例工厂，针对高并发的业务场景, 就使用SimpleRabbitListener即可
     */
    @Bean(name = "multiListenerContainer")
    public SimpleRabbitListenerContainerFactory multiListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        //设置连接工厂，可以使用configure对象配置
        factoryConfigurer.configure(factory,connectionFactory);
        factory.setMessageConverter(new Jackson2JsonMessageConverter());

        //消息的确认消费模式，手动确认消费
//        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        //高并发并发消费者实例最大数量  初始状态
        factory.setConcurrentConsumers(10);
        //并发的最大数量
        factory.setMaxConcurrentConsumers(15);
        //并发消费者实例每个实例拉取的消息数量
        factory.setPrefetchCount(10);

        return factory;
    }
    /**
     * 配置RabbitMQ发送消息组件RabbitTemplate
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        //设置发送消息后进行确认, connectionFactory的配置都直接在配置文件配置
//        connectionFactory.setPublisherConfirms(true);
//        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        RabbitTemplate rabbitTemplate = new RabbitTemplate();

        //生产者通过调用channel.addReturnListener()方法来添加ReturnListener监听器，实现获取没有被正确路由到合适队列的消息
        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setConnectionFactory(connectionFactory);
        //发消息如果成功，给出反馈信息
        rabbitTemplate.setConfirmCallback((correlationData, b, s) -> log.info("消息发送成功：correlationData({}),ack({}),cause({})",correlationData,b,s));

        //发送消息失败，给出反馈
        rabbitTemplate.setReturnsCallback(returnedMessage -> log.warn("消息丢失：exchange({}),route({}),replyCode({}),replyText({}),message({})",returnedMessage.getExchange(),returnedMessage.getRoutingKey(), returnedMessage.getReplyCode(),returnedMessage.getReplyText(),returnedMessage.getMessage()));

        return rabbitTemplate;
    }

    /**
     * 配置监听容器
     * 消费者监听实例
     */
    @Resource
    private BasicConsumer basicConsumer;

    @Bean(name = "simpleContainerManual")
    public SimpleMessageListenerContainer simpleMessageListenerContainer(@Qualifier("basicQueue") Queue basicQueue) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //消费实例配置
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setPrefetchCount(1);

        //消息确认模式
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        container.setQueues(basicQueue);

        //手动确认消费 消费者需要实现ChannelAwareMessageListener接口
        container.setMessageListener(basicConsumer);

        return container;
    }

    /**
     * 配置简单的消息模型 ：队列、路由、交换机
     */
    @Bean(name = "basicQueue")
    public Queue basicQueue() {
        return new Queue(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.queue-name")),true);
    }

    @Bean
    public DirectExchange basicExchange() {
        return new DirectExchange(environment.getProperty("spring.rabbitmq.exchange-name"),true,false);
    }

    @Bean
    public Binding basicBinding() {
        return BindingBuilder.bind(basicQueue()).to(basicExchange()).with(environment.getProperty("spring.rabbitmq.route-name"));
    }

    /**
     * 配置简单类型的消息模型： 对象类型   ： 队列、路由、交换机
     */
    @Bean(name = "objectQueue")
    public Queue objectQueue() {
        return new Queue(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.queue-object-name")));
    }

    @Bean
    public  DirectExchange objectExchange() {
        return new DirectExchange(environment.getProperty("spring.rabbitmq.exchange-object-name"),true,false);
    }

    @Bean
    public Binding objectBinding() {
        return BindingBuilder.bind(objectQueue()).to(objectExchange()).with(environment.getProperty("spring.rabbitmq.route-object-name"));
    }

    /**
     * 广播模式交换机、 不需要配置路由信息，因为路由无用，子网广播
     */
    @Bean(name = "fanoutQueueOne")
    public Queue fanoutQueueOne() {
        return new Queue(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.queue-one-name")));
    }

    @Bean(name = "fanoutQueueTwo")
    public Queue fanoutQueueTwo() {
        return new Queue(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.queue-two-name")));
    }

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.exchange-fanout-name")),true,false);
    }

    /**
     * 只需要将交换机和队列绑定再一起即可，不需要指定Bindinging Key
     */
    @Bean
    public Binding fanoutBindingOne() {
        return BindingBuilder.bind(fanoutQueueOne()).to(fanoutExchange());
    }

    @Bean
    public Binding fanoutBindingTwo() {
        return BindingBuilder.bind(fanoutQueueTwo()).to(fanoutExchange());
    }


    /**
     * 死信队列消息模型
     */
    @Bean
    public Queue basicDeadQueue() {
        //死信队列，需要指定Map
        Map<String,Object> arguments = new HashMap<>();
        //创建死信交换机
        arguments.put("x-dead-letter-exchange",Objects.requireNonNull(environment.getProperty("spring.rabbitmq.exchange-dead-name")));
        //拆功能键死信路由
        arguments.put("x-dead-letter-routing-key",Objects.requireNonNull(environment.getProperty("spring.rabbitmq.route-dead-name")));
        //设置TTL, 10 000ms， 10s
        arguments.put("x-message-ttl",10000);

        //创建死信队列
        return new Queue(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.queue-dead-name")),true,false,false,arguments);
    }

    @Bean
    public TopicExchange basicProducerExchange() {
        return new TopicExchange(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.exchange-basic-name")),true,false);
    }

    // 创建基本交换机和死信队列的绑定，并指定binding key
    @Bean
    public Binding basicProducerBinding() {
        return BindingBuilder.bind(basicDeadQueue()).to(basicProducerExchange()).with(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.route-basic-name")));
    }

    //消费者消费的真正队列
    @Bean
    public Queue realConsumerQueue() {
        return new Queue(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.queue-basic-name")),true);
    }

    //上面的死信队列只是指定了死信交换机和死信路由的名称，还需要创建出来
    @Bean
    public TopicExchange basicDeadExchange() {
        return new TopicExchange(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.exchange-dead-name")),true,false);
    }

    @Bean
    public Binding basicDeadBinding() {
        return BindingBuilder.bind(realConsumerQueue()).to(basicDeadExchange()).with(Objects.requireNonNull(environment.getProperty("spring.rabbitmq.route-dead-name")));
    }

}
